package cn.itcast.czxy

import java.util.Properties

import cn.itcast.czxy.BD18.bean.{HBaseMeta, TagRule}
import cn.itcast.czxy.BD18.tools.{Common, HBaseDataSource}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * @author 红尘丶世界
 * @version v 1.0
 */
/*
GenderTag 用于性别标签的计算
 */
object GenderTag {


  //程序的入口
  def main(args: Array[String]): Unit = {
    //创建工具包对象
    val common: Common = new Common
    //1  创建SparkSQL对象
    //   用于读取mysql ， hbase等数据
    val spark: SparkSession = SparkSession.builder().appName("GenderTag").master("local[*]").getOrCreate()


    //2  连接mysql 数据库
    val url: String = "jdbc:mysql://bd001:3306/tags_new2?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val table: String = "tbl_basic_tag" //mysql数据表的表名
    val properties: Properties = new Properties

    //连接mysql
    val mysqlConn: DataFrame = spark.read.jdbc(url, table, properties)


    //引入隐式转换
    import spark.implicits._
    //引入java 和scala相互转换
    import scala.collection.JavaConverters._
    //引入sparkSQL的内置函数
    import org.apache.spark.sql.functions._


    //3  读取mysql数据库中的四级标签
    //   为读取hbase数据做准备
    //id=81
    //rule=    inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,gender
    val fourTagsDS: Dataset[Row] = mysqlConn.select("id", "rule").where("id=4")

    //周勇江 四级标签开发
    // 这个字符串读取数据不好用   inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,gender
    // 将上述数据转为样例类，以便于后面读取数据
    // 遍历四级标签数据fourTags

    //调用工具类切分rule
    val KVMap: Map[String, String] = common.ruleSplit(fourTagsDS, spark)

    println(KVMap)

    //开发toHBaseMeta方法     将KVMap 封装成为样例类HBaseMeta
    var hbaseMeta: HBaseMeta = (new HBaseDataSource).parseMeta(KVMap)
    // println( hbaseMeta.hbaseTable+"    "+ hbaseMeta.family+"    "+ hbaseMeta.selectFields)

    //4  读取mysql数据库中方的五级标签
    //   匹配性别
    val fiveTags: Dataset[Row] = mysqlConn.select("id", "rule").where("pid = 4")
    val fiveTagsList: List[TagRule] = fiveTags.map(row => {
      //把一行数据封装成样例类
      TagRule(
        row.getAs("id").toString.toInt,
        row.getAs("rule").toString
      )
    }).collectAsList() //将DataSet 转化成util.List[TagRule] 这个类型无法获取id和rule的数据
      .asScala.toList //需要隐式转换 import scala.collection.JavaConverters._


    //5  根据mysql数据中的四级标签的规则   读取hbase 数据
    //若使用hbase 客户端读取效率较慢，将hbase作为数据源，读取效率较快
    val hbaseDatas: DataFrame = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      // hbaseMeta.zkHosts就是 192.168.10.20
      .option("zkHosts", hbaseMeta.zkHosts)
      //HBaseMeta.ZKPORT=>"zkPort"
      //hbaseMeta.zkPort=>2181
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, hbaseMeta.hbaseTable)
      .option(HBaseMeta.FAMILY, hbaseMeta.family)
      .option(HBaseMeta.SELECTFIELDS, hbaseMeta.selectFields)
      .load()
    hbaseDatas.show(20)


    //6  标签匹配
    //    根据五级标签数据和hbase数据进行标签匹配  的到最终的标签
    //编写udf函数  输入是1，2  返回值82 ，83
    val GetTagId: UserDefinedFunction = udf((gender: String) => { //rule:就是HBase中的1  2  1  2
      //设置标签默认值0
      var id: Int = 0
      //遍历五级标签
      for (ruleOb <- fiveTagsList) { // fiveTagList=>  82      1，83      2
        //当用户数据的gender与五及标签的rule相等，那么返回五及标签的id
        if (gender == ruleOb.rule) {
          id = ruleOb.id
        }
      }
      id
    })

    //标签匹配
    val userTags: DataFrame = hbaseDatas.select('id as ("userId"), GetTagId('gender).as("tagsId"))
    userTags.show()


    //7  将最终的标签写入HBase
//    userTags.write.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
//      .option("zkHosts", hbaseMeta.zkHosts)
//      //HBaseMeta.ZKPORT=>"zkPort"
//      //HBaseMeta.zkPort=>2181
//      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
//      .option(HBaseMeta.HBASETABLE, "test")
//      .option(HBaseMeta.FAMILY, "detail")
//      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
//      .save()

  }

}
